//! # 管理接口模块
//!
//! 重新导出 rustcloud-admin 模块的管理功能，
//! 为 rustcloud-stream 提供统一的管理接口。
//! 
//! 通过这种方式保持向后兼容性，同时使用公共功能模块。

// 重新导出 rustcloud-admin 的管理功能
pub use rustcloud_admin::prelude::*;

// 为流处理定制的管理接口实现
use async_trait::async_trait;
use axum::{
    extract::{Path, Query, State},
    response::Json,
    routing::{get, post, delete},
    Router,
};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use crate::core::StreamResult;

/// 流处理管理API
pub struct StreamAdminApi {
    /// 底层管理API
    inner: rustcloud_admin::api::AdminApi,
    /// 流管理器
    stream_manager: std::sync::Arc<tokio::sync::RwLock<StreamManager>>,
}

impl StreamAdminApi {
    /// 创建新的流管理API
    pub fn new(
        inner: rustcloud_admin::api::AdminApi,
        stream_manager: std::sync::Arc<tokio::sync::RwLock<StreamManager>>,
    ) -> Self {
        Self {
            inner,
            stream_manager,
        }
    }

    /// 获取流管理路由
    pub fn routes(&self) -> Router {
        let manager = self.stream_manager.clone();
        
        Router::new()
            .route("/streams", get(list_streams))
            .route("/streams", post(create_stream))
            .route("/streams/:stream_id", get(get_stream))
            .route("/streams/:stream_id", delete(delete_stream))
            .route("/streams/:stream_id/start", post(start_stream))
            .route("/streams/:stream_id/stop", post(stop_stream))
            .route("/streams/:stream_id/status", get(get_stream_status))
            .route("/streams/:stream_id/metrics", get(get_stream_metrics))
            .route("/streams/:stream_id/health", get(get_stream_health))
            .with_state(manager)
    }
}

/// 流管理器
pub struct StreamManager {
    /// 活动的流
    streams: HashMap<String, StreamInfo>,
}

impl StreamManager {
    /// 创建新的流管理器
    pub fn new() -> Self {
        Self {
            streams: HashMap::new(),
        }
    }

    /// 创建流
    pub async fn create_stream(&mut self, request: CreateStreamRequest) -> StreamResult<StreamInfo> {
        let stream_id = request.stream_id.clone();
        
        if self.streams.contains_key(&stream_id) {
            return Err(crate::error::StreamError::Validation(
                format!("流 {} 已存在", stream_id)
            ));
        }

        let stream_info = StreamInfo {
            id: stream_id.clone(),
            name: request.name,
            description: request.description,
            config: request.config,
            status: StreamStatus::Created,
            created_at: chrono::Utc::now(),
            updated_at: chrono::Utc::now(),
            metrics: StreamMetrics::default(),
        };

        self.streams.insert(stream_id, stream_info.clone());
        Ok(stream_info)
    }

    /// 获取流信息
    pub fn get_stream(&self, stream_id: &str) -> Option<&StreamInfo> {
        self.streams.get(stream_id)
    }

    /// 列出所有流
    pub fn list_streams(&self) -> Vec<StreamInfo> {
        self.streams.values().cloned().collect()
    }

    /// 删除流
    pub fn delete_stream(&mut self, stream_id: &str) -> StreamResult<()> {
        if let Some(stream) = self.streams.get(stream_id) {
            if matches!(stream.status, StreamStatus::Running) {
                return Err(crate::error::StreamError::Validation(
                    "无法删除正在运行的流，请先停止流".to_string()
                ));
            }
        }

        self.streams.remove(stream_id)
            .ok_or_else(|| crate::error::StreamError::NotFound(
                format!("流 {} 不存在", stream_id)
            ))?;

        Ok(())
    }

    /// 启动流
    pub async fn start_stream(&mut self, stream_id: &str) -> StreamResult<()> {
        let stream = self.streams.get_mut(stream_id)
            .ok_or_else(|| crate::error::StreamError::NotFound(
                format!("流 {} 不存在", stream_id)
            ))?;

        if matches!(stream.status, StreamStatus::Running) {
            return Err(crate::error::StreamError::Validation(
                "流已经在运行中".to_string()
            ));
        }

        stream.status = StreamStatus::Running;
        stream.updated_at = chrono::Utc::now();
        
        tracing::info!("启动流 {}", stream_id);
        Ok(())
    }

    /// 停止流
    pub async fn stop_stream(&mut self, stream_id: &str) -> StreamResult<()> {
        let stream = self.streams.get_mut(stream_id)
            .ok_or_else(|| crate::error::StreamError::NotFound(
                format!("流 {} 不存在", stream_id)
            ))?;

        if matches!(stream.status, StreamStatus::Stopped) {
            return Err(crate::error::StreamError::Validation(
                "流已经停止".to_string()
            ));
        }

        stream.status = StreamStatus::Stopped;
        stream.updated_at = chrono::Utc::now();
        
        tracing::info!("停止流 {}", stream_id);
        Ok(())
    }
}

impl Default for StreamManager {
    fn default() -> Self {
        Self::new()
    }
}

/// 流信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamInfo {
    /// 流ID
    pub id: String,
    /// 流名称
    pub name: String,
    /// 流描述
    pub description: Option<String>,
    /// 流配置
    pub config: serde_json::Value,
    /// 流状态
    pub status: StreamStatus,
    /// 创建时间
    pub created_at: chrono::DateTime<chrono::Utc>,
    /// 更新时间
    pub updated_at: chrono::DateTime<chrono::Utc>,
    /// 流指标
    pub metrics: StreamMetrics,
}

/// 流状态
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
pub enum StreamStatus {
    /// 已创建
    Created,
    /// 运行中
    Running,
    /// 已停止
    Stopped,
    /// 错误
    Error,
}

/// 流指标
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamMetrics {
    /// 总消息数
    pub total_messages: u64,
    /// 成功处理消息数
    pub successful_messages: u64,
    /// 失败处理消息数
    pub failed_messages: u64,
    /// 平均处理时间（毫秒）
    pub avg_processing_time_ms: f64,
    /// 吞吐量（消息/秒）
    pub throughput_msg_per_sec: f64,
}

impl Default for StreamMetrics {
    fn default() -> Self {
        Self {
            total_messages: 0,
            successful_messages: 0,
            failed_messages: 0,
            avg_processing_time_ms: 0.0,
            throughput_msg_per_sec: 0.0,
        }
    }
}

/// 创建流请求
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct CreateStreamRequest {
    /// 流ID
    pub stream_id: String,
    /// 流名称
    pub name: String,
    /// 流描述
    pub description: Option<String>,
    /// 流配置
    pub config: serde_json::Value,
}

/// 流状态响应
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct StreamStatusResponse {
    /// 流ID
    pub stream_id: String,
    /// 流状态
    pub status: StreamStatus,
    /// 运行时间（秒）
    pub uptime_seconds: Option<u64>,
    /// 最后更新时间
    pub last_updated: chrono::DateTime<chrono::Utc>,
}

// API 处理函数

/// 列出所有流
async fn list_streams(
    State(manager): State<std::sync::Arc<tokio::sync::RwLock<StreamManager>>>,
) -> Json<Vec<StreamInfo>> {
    let manager = manager.read().await;
    Json(manager.list_streams())
}

/// 创建流
async fn create_stream(
    State(manager): State<std::sync::Arc<tokio::sync::RwLock<StreamManager>>>,
    Json(request): Json<CreateStreamRequest>,
) -> Result<Json<StreamInfo>, rustcloud_admin::api::ApiError> {
    let mut manager = manager.write().await;
    match manager.create_stream(request).await {
        Ok(stream_info) => Ok(Json(stream_info)),
        Err(e) => Err(rustcloud_admin::api::ApiError::BadRequest(e.to_string())),
    }
}

/// 获取流信息
async fn get_stream(
    Path(stream_id): Path<String>,
    State(manager): State<std::sync::Arc<tokio::sync::RwLock<StreamManager>>>,
) -> Result<Json<StreamInfo>, rustcloud_admin::api::ApiError> {
    let manager = manager.read().await;
    match manager.get_stream(&stream_id) {
        Some(stream_info) => Ok(Json(stream_info.clone())),
        None => Err(rustcloud_admin::api::ApiError::NotFound(
            format!("流 {} 不存在", stream_id)
        )),
    }
}

/// 删除流
async fn delete_stream(
    Path(stream_id): Path<String>,
    State(manager): State<std::sync::Arc<tokio::sync::RwLock<StreamManager>>>,
) -> Result<Json<serde_json::Value>, rustcloud_admin::api::ApiError> {
    let mut manager = manager.write().await;
    match manager.delete_stream(&stream_id) {
        Ok(()) => Ok(Json(serde_json::json!({"message": "流删除成功"}))),
        Err(e) => Err(rustcloud_admin::api::ApiError::BadRequest(e.to_string())),
    }
}

/// 启动流
async fn start_stream(
    Path(stream_id): Path<String>,
    State(manager): State<std::sync::Arc<tokio::sync::RwLock<StreamManager>>>,
) -> Result<Json<serde_json::Value>, rustcloud_admin::api::ApiError> {
    let mut manager = manager.write().await;
    match manager.start_stream(&stream_id).await {
        Ok(()) => Ok(Json(serde_json::json!({"message": "流启动成功"}))),
        Err(e) => Err(rustcloud_admin::api::ApiError::BadRequest(e.to_string())),
    }
}

/// 停止流
async fn stop_stream(
    Path(stream_id): Path<String>,
    State(manager): State<std::sync::Arc<tokio::sync::RwLock<StreamManager>>>,
) -> Result<Json<serde_json::Value>, rustcloud_admin::api::ApiError> {
    let mut manager = manager.write().await;
    match manager.stop_stream(&stream_id).await {
        Ok(()) => Ok(Json(serde_json::json!({"message": "流停止成功"}))),
        Err(e) => Err(rustcloud_admin::api::ApiError::BadRequest(e.to_string())),
    }
}

/// 获取流状态
async fn get_stream_status(
    Path(stream_id): Path<String>,
    State(manager): State<std::sync::Arc<tokio::sync::RwLock<StreamManager>>>,
) -> Result<Json<StreamStatusResponse>, rustcloud_admin::api::ApiError> {
    let manager = manager.read().await;
    match manager.get_stream(&stream_id) {
        Some(stream_info) => {
            let uptime_seconds = if matches!(stream_info.status, StreamStatus::Running) {
                Some((chrono::Utc::now() - stream_info.updated_at).num_seconds() as u64)
            } else {
                None
            };

            Ok(Json(StreamStatusResponse {
                stream_id: stream_info.id.clone(),
                status: stream_info.status.clone(),
                uptime_seconds,
                last_updated: stream_info.updated_at,
            }))
        }
        None => Err(rustcloud_admin::api::ApiError::NotFound(
            format!("流 {} 不存在", stream_id)
        )),
    }
}

/// 获取流指标
async fn get_stream_metrics(
    Path(stream_id): Path<String>,
    State(manager): State<std::sync::Arc<tokio::sync::RwLock<StreamManager>>>,
) -> Result<Json<StreamMetrics>, rustcloud_admin::api::ApiError> {
    let manager = manager.read().await;
    match manager.get_stream(&stream_id) {
        Some(stream_info) => Ok(Json(stream_info.metrics.clone())),
        None => Err(rustcloud_admin::api::ApiError::NotFound(
            format!("流 {} 不存在", stream_id)
        )),
    }
}

/// 获取流健康状态
async fn get_stream_health(
    Path(stream_id): Path<String>,
    State(manager): State<std::sync::Arc<tokio::sync::RwLock<StreamManager>>>,
) -> Result<Json<serde_json::Value>, rustcloud_admin::api::ApiError> {
    let manager = manager.read().await;
    match manager.get_stream(&stream_id) {
        Some(stream_info) => {
            let health_status = match stream_info.status {
                StreamStatus::Running => "healthy",
                StreamStatus::Created | StreamStatus::Stopped => "stopped",
                StreamStatus::Error => "critical",
            };

            Ok(Json(serde_json::json!({
                "stream_id": stream_info.id,
                "status": health_status,
                "last_check": chrono::Utc::now(),
                "details": {
                    "stream_status": stream_info.status,
                    "total_messages": stream_info.metrics.total_messages,
                    "error_rate": if stream_info.metrics.total_messages > 0 {
                        (stream_info.metrics.failed_messages as f64 / stream_info.metrics.total_messages as f64) * 100.0
                    } else {
                        0.0
                    }
                }
            })))
        }
        None => Err(rustcloud_admin::api::ApiError::NotFound(
            format!("流 {} 不存在", stream_id)
        )),
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_stream_manager_create_stream() {
        let mut manager = StreamManager::new();
        
        let request = CreateStreamRequest {
            stream_id: "test-stream".to_string(),
            name: "测试流".to_string(),
            description: Some("这是一个测试流".to_string()),
            config: serde_json::json!({"input": "topic1", "output": "topic2"}),
        };

        let result = manager.create_stream(request).await;
        assert!(result.is_ok());

        let stream_info = result.unwrap();
        assert_eq!(stream_info.id, "test-stream");
        assert_eq!(stream_info.name, "测试流");
        assert_eq!(stream_info.status, StreamStatus::Created);
    }

    #[tokio::test]
    async fn test_stream_manager_start_stop_stream() {
        let mut manager = StreamManager::new();
        
        let request = CreateStreamRequest {
            stream_id: "test-stream".to_string(),
            name: "测试流".to_string(),
            description: None,
            config: serde_json::json!({}),
        };

        manager.create_stream(request).await.unwrap();

        // 启动流
        let result = manager.start_stream("test-stream").await;
        assert!(result.is_ok());

        let stream = manager.get_stream("test-stream").unwrap();
        assert_eq!(stream.status, StreamStatus::Running);

        // 停止流
        let result = manager.stop_stream("test-stream").await;
        assert!(result.is_ok());

        let stream = manager.get_stream("test-stream").unwrap();
        assert_eq!(stream.status, StreamStatus::Stopped);
    }

    #[test]
    fn test_stream_metrics_default() {
        let metrics = StreamMetrics::default();
        assert_eq!(metrics.total_messages, 0);
        assert_eq!(metrics.successful_messages, 0);
        assert_eq!(metrics.failed_messages, 0);
        assert_eq!(metrics.avg_processing_time_ms, 0.0);
        assert_eq!(metrics.throughput_msg_per_sec, 0.0);
    }
}